[#12453] improvement(core): add OCC for schema writes - #12456
Conversation
9abb06d to
531ca89
Compare
Code Coverage Report
Files |
0148763 to
0a61936
Compare
### What changes were proposed in this pull request? Add database-backed optimistic concurrency control and transaction boundaries for catalog writes. - Advance the catalog OCC version on every alter and on an overwrite insert, and guard alter and drop with a compare-and-set on the observed version, classifying a failed CAS as either a stale-version conflict or an already-missing entity. - Protect catalog creation with a shared lock on the parent metalake row on MySQL and PostgreSQL, without changing the parent version. H2 uses an exclusive lock because it has no shared row-lock syntax, so catalog creations under one metalake serialize on H2. - Keep the catalog CAS and the non-empty check or the cascade cleanup in one database transaction, and CAS-delete descendant schemas with their observed identifier-and-version pairs. - Keep the drop idempotent when another server deletes the catalog mid-request: a later store read such as listing schemas reports the catalog as missing, and the cached catalog wrapper is discarded. (`store.delete` maps a missing entity to `false` on its own.) - Comment the concurrency-critical statements, so the reason for their ordering is readable next to the code. Rebased on current `main` (on top of #12374). Second of three PRs replacing #12350. **Known window:** until #12456 lands, a schema created concurrently with a catalog drop can still slip through, because schema writes do not yet take the parent catalog row lock. #12456 adds that lock, which closes the window for both the cascade cleanup and the non-empty check. ### Why are the changes needed? Managed catalog operations previously consisted of multiple independent reads and writes. Concurrent alter, create, and drop requests could overwrite newer metadata, create a catalog below a metalake that was being deleted, or run partial cascade cleanup. Fix: #12452 ### Does this PR introduce _any_ user-facing change? Concurrent catalog version conflicts are reported as HTTP 409. If the observed entity was deleted or renamed away, alter reports not found and drop preserves its idempotent false result. ### How was this patch tested? - `./gradlew :core:test :core:javadoc -PskipITs` (H2) - New tests in `TestCatalogMetaService`, `TestCatalogManager`, `TestPOConverters`, including `testOverwriteInsertAdvancesCurrentVersion`, which checks that an overwrite moves the version forward and that a writer holding the pre-overwrite version no longer passes its CAS. - MySQL and PostgreSQL coverage is left to CI (`-PskipDockerTests=false`). --------- Co-authored-by: Jerry Shao <jerryshao@datastrato.com>
Advance the schema OCC version on every alter and guard alter and drop with a compare-and-set on the observed version, classifying a failed CAS as either a stale conflict or a missing entity. Make managed schema creation insert-only so a concurrent same-name create returns SchemaAlreadyExistsException instead of overwriting the winner, and take a shared lock on the parent catalog row so a schema cannot be created below a catalog that is being dropped. Serialize hierarchical ancestor materialization and schema drops through the catalog row so overlapping cascades share one lock order. Lock the parent schema row before writing a table, view, fileset, function, model, or topic, and check views and functions before a non-cascade schema drop. Accepted tradeoff: a hierarchical schema create that materializes implicit ancestors takes an exclusive lock on the catalog row, because two concurrent creates can both find the same ancestor missing and both insert it, and a shared lock does not prevent that under MySQL REPEATABLE READ.
… backends H2 is also the default embedded backend, not only a test backend. Spell out that falling back to an exclusive lock serializes schema creations under one catalog there and can surface as an H2 lock timeout.
…n code Review feedback: the concurrency-critical parts need comments so a reader can follow why the statements are ordered the way they are. - Say what the catalog row lock buys on a schema create, and why a nested name has to take it exclusively while a plain name does not. - Say why both drop paths delete the schema row before looking at its children, and why every drop takes catalog before schema. - Say what the shared schema lock in front of a table, view, fileset, function, model, or topic write is for, and that only a cross-schema rename needs it. - Say why the alter UPDATE compares only the version, what zero affected rows can mean, and why a partial cascade must roll back. - Say why managed schema creation is insert-only now. - Correct the schemaWriteFailure comment: sessions run at READ_COMMITTED, so the locking read is there to wait out an in-flight writer.
…ic on overwrite Carry the fix that apache#12455 already made for catalogs over to schemas, so both sides of the hierarchy follow the same rule. - Advance current_version on all four schema upsert paths (single and batch, on MySQL/H2 and PostgreSQL) instead of writing the initial version back, which would let a writer holding an older version still pass its own version check. - Name the table on the PostgreSQL assignments: a bare column on that side of ON CONFLICT is ambiguous there, which is how the previous CI run broke. - Add TestSchemaMetaPostgreSQLProvider to pin both rules without a database, so they are checked on every run and not only in the Docker-backed CI job. - Cover the race this PR is meant to close: a catalog cascade that holds the catalog row makes a concurrent schema create wait and then report the catalog as missing, leaving no orphan behind. Reading the cascade snapshot moved into a package-private method so the test can pause exactly at that point, the same seam MetalakeMetaService already offers. - Say that the H2 shared-lock fallback affects H2 backends, not just tests.
0a61936 to
cc97151
Compare
There was a problem hiding this comment.
Pull request overview
Adds database-backed optimistic concurrency control (OCC) and explicit row-lock/transaction fencing around managed schema writes to prevent lost updates, orphaned children, and deadlocks under concurrent create/alter/drop operations.
Changes:
- Advance schema OCC version on updates and enforce version compare-and-set semantics for alter and drop.
- Introduce catalog-row locks for schema creation/deletion (shared vs exclusive depending on hierarchical ancestor materialization) and schema-row shared locks for child entity inserts.
- Update managed schema create to be insert-only (no overwrite) and expand/restore concurrency-focused tests across core + catalog modules.
Reviewed changes
Copilot reviewed 24 out of 24 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java | Adds catalog/schema locking, CAS-based alter/drop, descendant handling, and empty-check extensions (views/functions). |
| core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java | Ensures schema updates always advance OCC version (current/last stay aligned). |
| core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java | Locks parent schema row before table insert; locks new parent schema on cross-schema rename. |
| core/src/main/java/org/apache/gravitino/storage/relational/service/ViewMetaService.java | Locks parent schema row before view insert to block writes during schema drop. |
| core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java | Locks parent schema row before topic insert to block writes during schema drop. |
| core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java | Locks parent schema row before fileset insert to block writes during schema drop. |
| core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java | Locks parent schema row before function insert to block writes during schema drop. |
| core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java | Locks parent schema row before model insert to block writes during schema drop. |
| core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java | Extracts schema snapshot read into an overrideable method to support concurrency tests. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java | Adds schema row lock selects and version-guarded soft delete mapper method. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java | Adds provider hooks for schema row locking + version-guarded delete; H2 share-lock fallback. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java | Implements FOR UPDATE / share-lock SQL and version-guarded soft delete; simplifies update WHERE to version CAS. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java | Adds PostgreSQL FOR SHARE and fixes ON CONFLICT version advancement with qualified columns. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java | Adds PostgreSQL FOR SHARE support for catalog row locking. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java | Adds share-lock SQL for catalog row. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java | Adds H2 share-lock fallback for catalog row locking. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java | Exposes catalog “FOR SHARE” select. |
| core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java | Makes managed schema create insert-only and maps “already exists” to SchemaAlreadyExistsException. |
| core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java | Updates expectations to verify schema version advancement on update conversion. |
| core/src/test/java/org/apache/gravitino/storage/relational/service/TestSchemaMetaService.java | Adds/extends concurrency and version-CAS tests for schema create/alter/drop and catalog-drop races. |
| core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java | Restores/adds cross-entity concurrency tests involving cascade delete vs concurrent schema ops. |
| core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestSchemaMetaPostgreSQLProvider.java | New test verifying PostgreSQL ON CONFLICT version columns are qualified and advanced. |
| catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java | Removes static meta-service mocking and uses real persisted metalake/catalog fixtures. |
| catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java | Adjusts schema creation helpers/fixtures to align with insert-only schema create behavior. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
|
I have no further comments about this PR. @yuqi1129 can you let AI to help to review for several rounds? |
I see. |
jerryshao
left a comment
There was a problem hiding this comment.
Automated review (Claude Code) — 1 confirmed correctness issue and 2 plausible efficiency/simplification items. See inline comments.
| // written below a schema that is being dropped. | ||
| () -> | ||
| SchemaMetaService.getInstance() | ||
| .lockSchemaForEntityWrite( |
There was a problem hiding this comment.
[Confirmed bug, surfaces at FilesetCatalogOperations.java:575] When this lock finds the schema gone (concurrently dropped), lockSchemaForEntityWrite throws NoSuchEntityException. But the call site in FilesetCatalogOperations.createMultipleLocationFileset's final store.put(filesetEntity, true) only catches IOException, so this exception propagates uncaught — past the earlier schema-existence pre-check in that same method, which correctly catches NoSuchEntityException and maps it to NoSuchSchemaException (404).
Failure scenario: Client A calls createFileset under schema S; the up-front store.get(schemaIdent,...) pre-check succeeds, then time passes doing filesystem mkdir/validation work. Meanwhile client B cascade-drops schema S. When A reaches store.put(filesetEntity, true), this lock finds the schema gone and throws NoSuchEntityException, which isn't NotFoundException-derived and isn't special-cased in ExceptionHandlers.java, so doWithCatalog lets it through unwrapped and the REST layer returns a raw 500 instead of the intended 404. The analogous managed-table path, ManagedTableOperations.createTable, does catch NoSuchEntityException and convert it to NoSuchSchemaException, showing this is the intended pattern the fileset call site misses.
There was a problem hiding this comment.
Confirmed, and I have updated the PR.
| * missing here, which let a non-cascade drop leave their rows behind with no parent. | ||
| */ | ||
| private void checkSchemaIsEmpty(NameIdentifier identifier, SchemaPO schemaPO) { | ||
| boolean hasDescendantSchemas = !listDescendantSchemaPOs(schemaPO).isEmpty(); |
There was a problem hiding this comment.
[Plausible efficiency] checkSchemaIsEmpty materializes six full PO lists (tables, filesets, models, topics, views, functions) via listXxxPOsBySchemaId just to test .isEmpty(), while holding an exclusive catalog-row lock the whole time.
A non-cascade drop of a schema with thousands of tables fetches and deserializes every table/fileset/model/topic/view/function row under that schema (six separate full scans) purely to answer a yes/no question, all while every other schema-create/drop under the same catalog is blocked on the exclusive catalog lock taken earlier in the same transaction. An existence check (COUNT/LIMIT 1 or an exists-by-schema-id query) would answer the same question with a fraction of the I/O and shorten how long the catalog stays locked.
There was a problem hiding this comment.
Yeah, It's indeed a performance problem. I would not modify it as this PR is already too large, and fixing it will introuce at least 300 lines of code changes. Moreover, the problem has existed for a long time; I would suggest fixing it in another PR.
| Long observedSchemaId, | ||
| Long observedCatalogId, | ||
| Long observedMetalakeId) { | ||
| NameIdentifier schemaIdentifier = NameIdentifierUtil.getSchemaIdentifier(entityIdentifier); |
There was a problem hiding this comment.
[Plausible simplification] The "lock the parent schema row before writing a child entity" pattern (this method) is hand-duplicated near-identically in six call sites (FilesetMetaService, FunctionMetaService, ModelMetaService, ModelVersionMetaService, TableMetaService, TopicMetaService, ViewMetaService) instead of living once in a shared write-path hook.
A future PR adding a seventh schema-scoped entity type could copy an existing insert method without also adding the SchemaMetaService.getInstance().lockSchemaForEntityWrite(...) call — nothing in the type system or a shared base class forces it, so the new entity type could silently reintroduce the exact orphan-under-a-dropped-schema race this PR was written to close for every other type.
There was a problem hiding this comment.
This is also a performance improvement, and I have created an issue: #12576 to track it.
### What changes were proposed in this pull request? - Use `current_version` as the OCC token for table alter and direct delete. - Execute the root table CAS before version, column, and dependent-row writes in one transaction. - Distinguish a concurrent version change from a deleted, renamed, or moved table. - Let optimistic-lock conflicts reach the caller on the managed paths, which write to the store directly. The best-effort `OperationDispatcher.operateOnEntity` helper keeps swallowing them: only external entities reach it, and there the catalog was already changed, so failing would invite a retry that re-applies a non-idempotent change. - Retry only optimistic-lock conflicts for idempotent Lance metadata repair. - Add comments explaining the transaction ordering, CAS predicates, conflict classification, and retry behavior. - Add comprehensive unit and cross-database tests. This PR depends on #12456. Two things are deliberately left out. The conflict classification (`tableWriteFailure`) and the locking select now exist once per entity type; extracting them is a follow-up once the series is done, because what differs per entity is the mapper, which identity columns to compare, and schema's physical-to-logical conversion, so doing it now means refactoring three merged services from inside a fourth. And a version conflict while dropping is tracked in #12597, which covers every entity type rather than this path alone. ### Why are the changes needed? Concurrent table writes could overwrite the winning version metadata, while a stale delete could remove data belonging to a newer table version. Some optimistic-lock conflicts were also swallowed or treated as generic IO failures. Fix: #12345 ### Does this PR introduce _any_ user-facing change? Stale managed-table writes now fail with the existing optimistic-lock conflict response instead of silently overwriting newer metadata. Reads and the import path are unchanged: loading an entity still repairs the Gravitino copy on a best-effort basis, so a load that loses a version race keeps returning the entity. No API or property is added. ### How was this patch tested? - Ran 48 `TableMetaService` cases across H2, MySQL, and PostgreSQL. - Ran all 16 `TestTableOperationDispatcher` tests. - Ran 4 SQL provider OCC tests. - Ran 27 Lance table and concurrent-repair tests. - Ran Spotless and `git diff --check`.
What changes were proposed in this pull request?
Add database-backed optimistic concurrency control and transaction boundaries for schema writes.
Accepted tradeoff: a hierarchical schema create that materializes implicit ancestors takes an exclusive lock on the catalog row, so every other schema create under that catalog waits until that transaction ends, even when it touches a different ancestor path. The exclusive lock is needed because two concurrent creates can both find the same ancestor missing and both insert it, and a shared lock does not prevent that under MySQL REPEATABLE READ. Catalogs with heavy concurrent hierarchical schema creation will therefore serialize on this lock. If it becomes a bottleneck, a narrower fence — locking only the ancestor rows being created and relying on the unique constraint plus a retry — can be done in a follow-up.
Rebased on current
main(on top of #12374). Third of three PRs replacing #12350. Stacked on the catalog PR; review the top commit only. This PR also restores the two cross-entityTestMetalakeMetaServicetests that could not pass before schema writes took the catalog row lock.Why are the changes needed?
Managed schema operations previously consisted of multiple independent reads and writes. Concurrent alter, create, and drop requests could overwrite newer metadata, create children below a deleted parent, leave view and function rows orphaned, or run partial cascade cleanup. Overlapping hierarchical schema drops could also acquire descendant row locks in different orders.
Fix: #12453
Does this PR introduce any user-facing change?
Concurrent schema version conflicts are reported as HTTP 409. If the observed entity was deleted or renamed away, alter reports not found and drop preserves its idempotent false result. A managed schema create that loses a concurrent same-name create returns
SchemaAlreadyExistsExceptioninstead of overwriting the winner.How was this patch tested?
./gradlew :core:test :core:javadoc :catalogs:catalog-fileset:test :catalogs:catalog-kafka:test -PskipITs(H2)TestSchemaMetaService,TestMetalakeMetaService,TestFilesetCatalogOperations,TestKafkaCatalogOperations.-PskipDockerTests=false).